package net.xdclass.xdclassmq.jms.consumer;

import net.xdclass.xdclassmq.config.RocketMqConfig;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component;

/**
 * @Author whg
 * @create 2023/1/3 10:49
 * @description：
 */
@Component
public class OrderConsumer {

    @Autowired
    private RocketMqConfig rocketMqConfig;

    private DefaultMQPushConsumer consumer;

    public OrderConsumer(){}

    @Bean
    public void OrderConsumer() throws MQClientException {
        //实例化consumer
        consumer = new DefaultMQPushConsumer(rocketMqConfig.getConsumerGroup());
        //设置nameserver地址
        consumer.setNamesrvAddr(rocketMqConfig.getNameServerAddr());
        //设置消费消息起始位置
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
        //设置消费指定tag的消息
        consumer.subscribe(rocketMqConfig.getOrderTopic(),"*");
        //默认是集群方式，可以更改为广播，但是广播方式不支持重试
        consumer.setMessageModel(MessageModel.BROADCASTING);
        //设置监听器
        consumer.registerMessageListener((MessageListenerConcurrently)(messages,context) -> {
            int times = 0;
            try{
                //Message msg = messages.get(0);
                MessageExt msg = messages.get(0);
                times = msg.getReconsumeTimes();
                System.out.println("消费重试次数：" + times);
                //消息体
                String body = new String(msg.getBody(),"utf-8");
                String topic = msg.getTopic();
                String tags = msg.getTags();
                String keys = msg.getKeys();
                if(keys.equals("6688")){
                    throw new Exception();
                }
                System.out.printf("%s Receive New Messages: %s %n",
                        Thread.currentThread().getName(),
                        body);
                System.out.println("topic=" + topic + ", tags=" + tags + ", keys=" + keys + ", msg=" + body);
                //返回 消费成功 给rocketmq
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }catch (Exception e){
                System.out.println("消息异常");
                //如果重试两次不成功，则记录，人工介入
                if(times >= 2){
                    //todo 记录数据库，发短信通知开发人员或运维人员
                    System.out.println("重试次数大于等于2，记录数据库，发短信通知开发人员或运维人员");
                    //告知broker，消费成功
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
                e.printStackTrace();
                //返回 稍后消费 给rocketmq，稍后会重新消费该消息
                return ConsumeConcurrentlyStatus.RECONSUME_LATER;
            }
        });
        consumer.start();
    }
}
